package com.ali.demo.unicom.impl;

import java.util.Date;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.transaction.support.TransactionTemplate;

import com.ali.demo.unicom.dao.DemoDAO;
import com.ali.demo.unicom.domain.Demo;
import com.ali.demo.unicom.exception.DemoException;
import com.ali.demo.unicom.ons.ONSHelper;
import com.ali.demo.unicom.ons.ONSTransactionBean;
import com.ali.demo.unicom.service.DemoService;
import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.exception.ONSClientException;
import com.aliyun.openservices.ons.api.transaction.LocalTransactionChecker;
import com.aliyun.openservices.ons.api.transaction.LocalTransactionExecuter;
import com.aliyun.openservices.ons.api.transaction.TransactionStatus;

public class DemoServiceImpl implements DemoService, MessageListener,
	LocalTransactionChecker {

    private static final Logger logger = LoggerFactory
	    .getLogger(DemoServiceImpl.class);

    private ONSTransactionBean transactionBean;
    private TransactionTemplate transactionTemplate;
    private DemoDAO demoDAO;

    public ONSTransactionBean getTransactionBean() {
	return transactionBean;
    }

    public void setTransactionBean(ONSTransactionBean transactionBean) {
	this.transactionBean = transactionBean;
    }

    public TransactionTemplate getTransactionTemplate() {
	return transactionTemplate;
    }

    public void setTransactionTemplate(TransactionTemplate transactionTemplate) {
	this.transactionTemplate = transactionTemplate;
    }

    public DemoDAO getDemoDAO() {
	return demoDAO;
    }

    public void setDemoDAO(DemoDAO demoDAO) {
	this.demoDAO = demoDAO;
    }

    @Override
    public long createDemo(final Demo demo) {
	try {
	    final long demoID = demoDAO.generateID();
	    demo.setCreateDate(new Date());
	    demo.setId(demoID);

	    Message msg = new Message();
	    msg.setTopic("demo_topic");
	    msg.setTag("DEMO");
	    msg.setKey(String.valueOf(demoID));
	    msg.setBody(ONSHelper.obj2Body(demo));

	    SendResult sendResult = transactionBean.send(msg,
		    new LocalTransactionExecuter() {

			@Override
			public TransactionStatus execute(Message msg, Object arg) {
			    try {
				demoDAO.insert(demo);
			    } catch (DataAccessException e) {
				String errmsg = "Demo insert failed: ID = "
					+ demoID + ", msg = " + msg.getMsgID();
				logger.error(errmsg, e);
				throw e;
			    }
			    return TransactionStatus.CommitTransaction;
			}
		    }, null);
	    if (logger.isInfoEnabled()) {
		logger.info("Demo created: ID = " + demoID + ", sent msg = "
			+ sendResult.getMessageId());
	    }

	    return demoID;
	} catch (DataAccessException e) {
	    String errmsg = "Subject create failed: " + e.getMessage();
	    logger.error(errmsg, e);
	    throw new DemoException(errmsg, e);
	} catch (ONSClientException e) {
	    String errmsg = "Send CREATE message failed: " + e.getMessage();
	    logger.error(errmsg, e);
	    throw new DemoException(errmsg, e);
	}
    }

    @Override
    public void updateDemo(Demo demo) {
	try {
	    demoDAO.update(demo);
	} catch (DataAccessException e) {
	    String errmsg = "Demo update failed: " + e.getMessage() + ", ID = "
		    + demo.getId();
	    logger.error(errmsg, e);
	    throw new DemoException(errmsg, e);
	}
    }

    @Override
    public Demo getDemoByID(final long demoID) {
	try {
	    return demoDAO.selectByID(demoID);
	} catch (DataAccessException e) {
	    String errmsg = "Demo load failed: " + e.getMessage() + ", ID = "
		    + demoID;
	    logger.error(errmsg, e);
	    throw new DemoException(errmsg, e);
	}
    }

    @Override
    public TransactionStatus check(Message msg) {
	if ("DEMO".equals(msg.getTag())) {
	    final long demoID = Long.parseLong(msg.getKey());
	    try {
		Demo demo = demoDAO.selectByID(demoID);
		if (demo != null) {
		    return TransactionStatus.CommitTransaction;
		}
		return TransactionStatus.RollbackTransaction;

	    } catch (DataAccessException e) {
		String errmsg = "Checking message failed: " + e.getMessage()
			+ ", ID = " + demoID + ", msg = " + msg.getMsgID();
		logger.error(errmsg, e);
		return TransactionStatus.Unknow;
	    }
	}

	return TransactionStatus.RollbackTransaction;
    }

    @Override
    public Action consume(Message msg, ConsumeContext context) {

	if ("DEMO".equals(msg.getTag())) {
	    Demo demo = ONSHelper.body2Obj(msg.getBody(), Demo.class);
	    if (logger.isInfoEnabled()) {
		String infomsg = "Demo insert: ID = " + msg.getKey()
			+ ", object = " + demo;
		logger.info(infomsg);
	    }
	    return Action.CommitMessage;

	} else {
	    String errmsg = "Unknown message TAG: " + msg.getTag() + ", KEY = "
		    + msg.getKey() + ", msgID = " + msg.getMsgID();
	    logger.error(errmsg);
	}

	return Action.ReconsumeLater;
    }
}
